package org.paraj.prodcons;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;


public abstract class BufferedConsumer<T> implements Consumer<T>{
    private Collection<T> pending = new ArrayList<T>();
    private long lastUpdate = System.currentTimeMillis();
    private long bufferSize;
    private long flushTimeout;
//    private Logger logger = Logger.getLogger(BufferedConsumer.class.getCanonicalName());

    public BufferedConsumer(long bufferSize, long flushTimeout){
        this.bufferSize = bufferSize;
        this.flushTimeout = flushTimeout;
    }

    public void consume(T product) throws Exception{
        pending.add(product);
        if (pending.size() >= bufferSize || System.currentTimeMillis() -  lastUpdate > flushTimeout){
            try{
                doConsume(pending);
            }catch(Throwable t){
                String msg = String.format("Error '%s' while processing the following group: %s", t.getMessage(), Arrays.toString(pending.toArray()));
                throw new RuntimeException( msg, t);
            } finally {
                lastUpdate = System.currentTimeMillis();
                pending.clear();                
            }
        }
    }

    protected abstract void doConsume(Collection<T> pending);

    public void close(){
        if (!pending.isEmpty()){
            doConsume(pending);
        }
    }
}
